#include "cc_port.h"
#include "cc_shell.h"

#if (CC_CONFIG_TEST_PORT == 1)

void test_listen_cb(e_cc_port src_id, e_cc_port dst_id, const char *topic, void* msg)
{
  cc_log("port(%d) recv from port(%d) topic(%s)\r\n", dst_id, src_id, topic);
  return;
}

void send_port_test(void *data)
{
  cc_port_t port = cc_port_init(CC_PORT_TEST1, "test1", 10, MEM_BLOCK_REAL_SIZE(50));
  cc_port_set_state(port, CC_PORT_STATE_CONN);

  int cnt = 0;

  while (1)
  {
    cc_mnode_t msg_node = cc_port_get_msg_node(port, 1000);
    sprintf((char *)msg_node->mem, "hello world");
    if (cc_port_send(port, CC_PORT_TEST2, msg_node) == CC_EOK)
    {
      cc_log_d("send to port(%d) success\r\n", CC_PORT_TEST2);
    }

    msg_node = cc_port_get_msg_node(port, 1000);
    sprintf((char *)msg_node->mem, "I am coming test4");
    if (cc_port_send(port, CC_PORT_TEST4, msg_node) == CC_EOK)
    {
      cc_log_d("send to port(%d) success\r\n", CC_PORT_TEST4);
    }

    if (cnt++ == 5)
    {
      cc_mnode_t broad_cast_node = cc_port_get_msg_node(port, 1000);
      if (broad_cast_node)
      {
        cc_port_broadcast(port, broad_cast_node);
      }
      cc_mnode_t publish_node = cc_port_get_msg_node(port, 1000);
      cc_port_publish(port, "test_topic", publish_node);
      publish_node = cc_port_get_msg_node(port, 1000);
      cc_port_publish(port, "test_topic2", publish_node);
      cnt = 0;
    }
    cc_delay(1000);
  }
}
CC_THREAD_REGISTER(send_port_test, 10, 3096);

void recv_port_test(void *data)
{
  cc_port_t port = cc_port_init(CC_PORT_TEST2, "test2", 10, MEM_BLOCK_REAL_SIZE(50));
  cc_port_listen(port, test_listen_cb);
  cc_port_subsribe(port, "test_topic");
  cc_port_subsribe(port, "test_topic2");
  cc_port_set_state(port, CC_PORT_STATE_CONN);
  int cnt;
  while (1)
  {
    cc_mnode_t node = cc_port_recv(port);
    if (node)
    {
      char *data = (char *)node->mem;
      cc_log_d("recv success data:%s\r\n", data);
      cc_mnode_put(node);
    }
    cnt++;
    if (cnt == 10)
    {
      if (CC_EOK == cc_port_unsubsribe(port, "test_topic2"))
        cc_log("success unsub\r\n");
    }
    if (cnt == 15)
    {
      cc_port_disconnect(port);
    }
    cc_delay(1000);
  }
}
CC_THREAD_REGISTER(recv_port_test, 10, 3096);

void recv2_port_test(void *data)
{
  cc_delay(3000);
  cc_port_t port = cc_port_init(CC_PORT_TEST3, "test3", 10, MEM_BLOCK_REAL_SIZE(50));
  cc_port_listen(port, test_listen_cb);
  cc_port_subsribe(port, "test_topic");
  cc_port_subsribe(port, "test_topic");
  cc_port_set_state(port, CC_PORT_STATE_CONN);

  while (1)
  {
    cc_delay(1000);
  }
}
CC_THREAD_REGISTER(recv2_port_test, 10, 3096);

void port4_listen(e_cc_port src_id, e_cc_port dst_id,  const char *topic, void* msg)
{
  cc_log("port(%d) recv from port(%d) topic(%s)\r\n", dst_id, src_id, topic);
  if (msg && strcmp(topic, "publish") != 0)
  {
    cc_log("recv success data:%s\r\n", msg);
  }
  return;
}

void recv3_port_test(void *data)
{
  cc_delay(3000);
  cc_port_t port = cc_port_init(CC_PORT_TEST4, "test4", 10, MEM_BLOCK_REAL_SIZE(50));
  cc_port_listen(port, port4_listen);
  cc_port_set_state(port, CC_PORT_STATE_CONN);

  while (1)
  {
    if (cc_port_recv_listen(port) == CC_EOK)
    {
      cc_log_d("port(%d) recv success\r\n", CC_PORT_TEST4);
    }
    cc_delay(1000);
  }
}
CC_THREAD_REGISTER(recv3_port_test, 10, 3096);

#endif